package com.net.protocol;


import android.content.Context;

import com.net.protocol.exception.ConnectException;
import com.net.protocol.exception.IllegalMessageException;
import com.net.protocol.exception.IllegalOperateException;
import com.net.protocol.exception.InitException;
import com.net.protocol.message.Message;
import com.net.protocol.message.body.receiver.Search_ACK;
import com.net.protocol.util.Logger;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;


/**
 * TCP管理器，处理终端消息的发送及平台消息的接收。
 * 1.客户端将消息放入阻塞队列，然后进入wait状态，在传入的Message对象上等待唤醒(等待超时时间10s);
 * 2.服务端处理消息， 唤醒等待在对应Message对象上的客户端线程
 * 3.接收服务端下发的消息，通知给监听者
 *
 */
public class TCPManager implements TCPConnector.ConnectListener {

    /**
     * 消息接收监听
     */
    public interface MessageReceiveListener {
        /**
         * 消息接收的回调
         *
         * @param resp 回复
         */
        public void onReceive(Message<Message.Body.Receiver> resp);
    }

    private static final String TAG = "{TCPManager}";

    private static final String HOST = "0.0.0.0"; // 外网
    private static final int PORT = 13000; // 端口号

    private static final int OFFER_BLOCKING_QUEUE_TIMEOUT = 100; // 向存放消息的阻塞队列放入数据的超时时间
    private static final int HEARTBEAT_INTERVAL = 25000; // 心跳包发送间隔

    private volatile static TCPManager tcpManager; // TCP管理
    private TCPConnector tcpConnector; // TCP连接
    private MessageSender messageSender; // 发送消息的线程
    private MessageReceiver messageReceiver; // 接收消息的线程
    private Timer timer;

    private static byte[] sendBuffer = new byte[1024]; // 发送缓冲
    private static byte[] receiveBuffer = new byte[1024]; // 接收缓冲

    private static final BlockingQueue<Message<? extends Message.Body.Sender>> blockingQueue = new LinkedBlockingQueue<>(200); // 存放消息的阻塞队列
    private static final Map<Long, MessageReceiveListener> waittingMap = new ConcurrentHashMap<>(); // 存放处于等待处理的回调对象
    private static final List<TCPConnector.ConnectListener> connListener = Collections.synchronizedList(new ArrayList<TCPConnector.ConnectListener>()); // 存放业务连接状态的监听者

    private Context context;
    private int receiveOffset = 0; //读入数据的偏移量
    private int reqSequenceId; // 发送消息的序列号
    private long receivedSeqId; //最后接收到的有效流水号

    private long deviceId;

    /**
     * 私有化构造方法
     */
    private TCPManager(Context context) {
        this.context = context;
        init();
    }

    /**
     * 初始化
     */
    private void init() {
        tcpConnector = new TCPConnector(HOST, PORT, this, "business-connector");
        messageSender = new MessageSender("business-message-sender"); // 创建发送消息线程
        messageReceiver = new MessageReceiver("business-message-receiver"); // 创建接收消息线程
    }

    /**
     * 连接建立，连接建立后立即进行鉴权，鉴权成功后启动消息接收及补传线程
     */
    @Override
    public void onConnected() {
        Logger.i(TAG, "1.TCP连接成功,开始发送鉴权消息");
        messageReceiver.restart(); // 启动接收线程
        messageSender.restart(); // 启动补传线程
        if (timer == null) {
            timer = new Timer();
            timer.schedule(new HeartbeatTimerTask(), new Date(), HEARTBEAT_INTERVAL); //心跳包发送任务
        }
        notifyConnObserver(true); //通知监听者连接建立

    }

    /**
     * 连接断开
     */
    @Override
    public void onDisconnected() {
        Logger.i(TAG, "9.TCP连接断开，停止消息发送、接收、补发线程、心跳包发送任务");
        if (timer != null) {
            timer.cancel();
            timer = null;
        }
    }

    /**
     * 获取TCPManager实例
     */
    public static TCPManager getInstance(Context context) {
        if (tcpManager == null) {
            synchronized (TCPManager.class) {
                if (tcpManager == null) {
                    tcpManager = new TCPManager(context);
                }
            }
        }
        return tcpManager;
    }

    /**
     * 获取TCPManager实例
     */
    public static TCPManager getInstance() throws IllegalOperateException {
        if (tcpManager == null) {
            Logger.e(TAG, "该方法只能在getInstance(Context context)之后调用");
            throw new IllegalOperateException("该方法只能在getInstance(Context context)之后调用");
        }
        return tcpManager;
    }

    /**
     * 建立业务连接
     */
    private void connect() {
        if (tcpConnector.isConnected()) {
            return;
        }
        tcpConnector.startRunning(); //启动业务连接
    }

    /**
     * 启动TCP业务链接
     */
    public void start(){
        connect();
    }

    /**
     * 停止业务链接,建立休眠连接
     */
    public void stop() throws IllegalOperateException, InitException {
        tcpConnector.stopRunning(); //停止TCP业务连接
        if (timer != null) {
            timer.cancel(); //取消心跳包定时发送任务
            timer = null;
        }
        notifyConnObserver(false); //通知监听者连接断开
    }

    /**
     * 通知监听者业务连接状态发生改变
     *
     * @param connected 连接建立或者断开
     */
    private void notifyConnObserver(boolean connected) {
        for (TCPConnector.ConnectListener listener : connListener) {
            if (connected) {
                listener.onConnected();
            } else {
                listener.onDisconnected();
            }
        }
    }

    /**
     * 获取消息流水号
     */
    private synchronized int getSeqId() {
        return ++reqSequenceId;
    }

    /**
     * 发送消息
     *
     * @param msg 消息
     * @return 消息发送结果
     * @throws IllegalMessageException
     * @throws InterruptedException
     */
    private int sendMessage(Message<? extends Message.Body.Sender> msg) throws IllegalMessageException, InterruptedException {
        int seqId = getSeqId();
        msg.getHeader().setSeqId(seqId); // 设置发送消息id

        blockingQueue.offer(msg, OFFER_BLOCKING_QUEUE_TIMEOUT, TimeUnit.MILLISECONDS); // 将数据放入阻塞队列
        return seqId;
    }

    /**
     * 发送消息
     *
     * @param msg      消息
     * @param listener 消息发送完成后的监听
     * @return 消息发送结果
     * @throws IllegalMessageException
     * @throws InterruptedException
     */
    public int sendASKMessage(Message<Message.Body.Sender.ASK> msg, MessageReceiveListener listener)
            throws IllegalMessageException, InterruptedException {
        int result = sendMessage(msg);

        if (listener != null) {
            waittingMap.put(msg.getHeader().getSeqId(), listener); // 消息等待处理,处理完成后将会通知监听者
        }
        return result;
    }

    /**
     * 发送消息
     *
     * @param msg 消息
     * @return 消息发送结果
     * @throws IllegalMessageException
     * @throws InterruptedException
     */
    public int sendACKMessage(Message<Message.Body.Sender.ACK> msg) throws IllegalMessageException, InterruptedException {
        return sendMessage(msg);
    }

    /**
     * 处理接收到的消息
     *
     * @param msg 接收到的消息
     */
    private void handleReceivedMessage(Message<Message.Body.Receiver> msg) {
        if (msg.getStatus() == Message.MESSAGE_STATUS_OK) {
            receivedSeqId = msg.getHeader().getSeqId(); //更新收到消息的流水号
            if (msg.getBody() instanceof Search_ACK) { //平台确认消息
                EventController.searchFileEvent(((Search_ACK) msg.getBody()).fileLists);
               //todo 判断不同的消息类型外发

            }
        }
    }

    /**
     * 心跳包定时发送任务
     */
    private class HeartbeatTimerTask extends TimerTask {

        private static final String TAG = "{TCPManager$HeartbeatTimerTask}";

        @Override
        public void run() {
            Logger.i(TAG, "====1.发送心跳包====");
            if (System.currentTimeMillis() - tcpConnector.getLastCommTime() > HEARTBEAT_INTERVAL) {
                Logger.i(TAG, "====1.2发送心跳包====");
                //test ----begin
                try {
                    tcpConnector.write("hello".getBytes(),0,"hello".getBytes().length);
                } catch (ConnectException e) {
                    e.printStackTrace();
                }
                //test ----end
//                try {
//                    sendASKMessage(MessageBuilder.build(new Heartbeat((byte) 0)), null);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                } catch (IllegalMessageException e) {
//                    e.printStackTrace();
//                }
            }
        }
    }

    /**
     * 发送消息的线程,不断从阻塞队列中取消息，当队列为空时线程处于阻塞状态，若取到消息，则将消息发送到TCP通道
     */
    private class MessageSender extends RestartableThread {

        private static final String TAG = "{TCPManager$MessageSender}";

        public MessageSender(String name) {
            this.setName(name);
        }

        @Override
        public void execute() {
            Logger.i(TAG, "====1.启动消息发送线程====");
            while (tcpConnector.isConnected()) {
                try {
                    Message msg = blockingQueue.take(); // 从消息队列头取消息，队列为空则阻塞
                    //Logger.i(TAG, "====2.将消息写入TCP通道====" + msg);
                    int length = msg.pack(sendBuffer); //对消息进行打包转义
                    tcpConnector.write(sendBuffer, 0, length); // 将消息发送到TCP通道
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (IllegalMessageException e) {
                    e.printStackTrace();
                } catch (ConnectException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    /**
     * 接收消息的线程，从TCP输入流中读取并解析数据
     */
    private class MessageReceiver extends RestartableThread {

        private static final String TAG = "{TCPManager$MessageReceiver}";

        public MessageReceiver(String name) {
            this.setName(name);
        }

        @Override
        public void execute() {
            Logger.i(TAG, "====1.启动消息接收线程====");
            while (tcpConnector.isConnected()) {
                Logger.i(TAG, "====2.开始从TCP通道读取消息====");
                try {
                    int length = tcpConnector.read(receiveBuffer, receiveOffset); // 从TCP通道读入缓冲数组中的数据长度
                    //test begin
                    String s = new String(receiveBuffer);
                    EventController.logTest(s);
                    //test end
//                    System.out.println("====3.将数据读入接收缓冲====[receiveOffset:" + receiveOffset + ",length:" + length + "]");
//                    receiveOffset = 0;
//                    int offset = 0; // 当前已处理数据在缓冲数组中的偏移量
//                    while (offset < length) {
//                        try {
//                            Message msg = Message.unpack(receiveBuffer, offset, length - offset); // 解包
//                            //Logger.i(TAG, "====4.解包====" + msg);
//                            if (msg.getStatus() == Message.MESSAGE_STATUS_OK) {
//                                handleReceivedMessage(msg); //处理接收到的消息
//                                offset += msg.getSize(); // 更新接收缓冲的偏移量，继续解包
//                            } else if (msg.getStatus() == Message.MESSAGE_STATUS_LACK) { // 消息不完整，继续读取
//                                tidyReceiveBuffer(offset, length - offset);
//                                break;
//                            }
//                        } catch (IllegalMessageException e) {
//                            e.printStackTrace();
//                            receiveOffset = 0;
//                            break;
//                        }
//                    }
                } catch (ConnectException e) {
                    e.printStackTrace();
                }
            }
        }

        /**
         * 整理接收缓冲
         *
         * @param offset 已经处理到的数据偏移量
         * @param length 剩余未处理数据长度
         */
        private void tidyReceiveBuffer(int offset, int length) {
            Logger.i(TAG, "====6.整理接收缓冲区====offset:" + offset + ",length:" + length);
            System.arraycopy(receiveBuffer, offset, receiveBuffer, 0, length); // 将剩余未处理数据拷贝到数组头部，然后从偏移位置开始读入数据
            receiveOffset = length;
        }



    }

    /**
     * 测试发送
     * @param name
     * @param pwd
     */
    public void testSend(String name,String pwd){
        try {
            tcpConnector.write((name+pwd).getBytes("GBK"),0,(name+pwd).getBytes("GBK").length);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
